package org.jxkj.app

import org.apache.spark.SparkContext
import org.apache.spark.sql.{DataFrame, SparkSession}
import org.jxkj.data.IDaoProvider
import org.jxkj.util.JzCondition
import java.text.SimpleDateFormat
import java.util.Date

/**
 * 1、一次调频考核计算
 * 2、AGC调节考核最新版
 */
object FirstFrequency {

  def max(i: String, j: String): String = {
    if (i.compareTo(j) < 0)
      j
    else
      i
  }

  def max(i: Double, j: Double) = if (i < j) j else i

  def min(i: String, j: String) = if (i.compareTo(j) < 0) i else j

  def min(i: Double, j: Double) = if (i < j) i else j


  // 一次调频（重构版20210726，使用高频数据来计算，最小维度是毫秒）
  def calFrequency2(ss: SparkSession, sc: SparkContext, hTable: IDaoProvider, time: Array[String], step: Int): Unit = {
    import ss.implicits._
    // 时间
    val startTime = time(0)
    val endTime = time(1)

    val sdf = new SimpleDateFormat("yyyy/MM/dd")
    val sdfs = new SimpleDateFormat("yyyy/MM/dd HH:mm:ss")

    val startTime2 = new Date(sdfs.parse(startTime).getTime)
    val endTime2 = new Date(sdfs.parse(endTime).getTime + 999)
    var time2 = Array(startTime2, endTime2)

    // 上下限
    val low = 50 - 0.05
    val up = 50 + 0.05
    val abc = 0.04
    // 持续时间的最小值（单位：毫秒）
    val during_time_param1 = 1000
    // 有效性能事件的持续时间最小值（单位：毫秒）
    val during_time_param2 = 10000
    // 两次事件间隔（单位：毫秒）
    val normal_interval_param1 = 8000
    // 免考核系数
    val power_rated_factor = 0.2


    // 从数据库配置表读取计算测点
    val jdbcDF = ss.read
      .format("jdbc")
      .option("url", MysqlDml.url)
      .option("dbtable", "hms_first_frequency_pmu_point_conf")
      .option("user", MysqlDml.user)
      .option("password", MysqlDml.password)
      .option("driver", MysqlDml.driver)
      .load()
      .cache()
    jdbcDF.createOrReplaceTempView("hms_first_frequency_pmu_point_conf")

    val pointDF = ss.sql("select * from hms_first_frequency_pmu_point_conf where category in ('power','freq','act') and point_name not like '00F%' ")
    pointDF.createOrReplaceTempView("tb_point")
    // 测点
    val point_code: Array[String] = pointDF.map(x => x.getString(0)).collect()


    // 从PMU高频数据取值，20ms一个数据
    GetData.getByMillisFromPmu(ss,sc, point_code, time2);

    // 每秒的值分组分类，power_rated —— 额定功率
    val odsDF = ss.sql(
      s"""
         | select
         |   substr(a.point_name, 1, 3) as unit_name,
         |   b.createTime as createTime,
         |   sum((case when a.category = 'power' then b.pvalue end)) as power,
         |   sum((case when a.category = 'freq' then b.pvalue end)) as freq,
         |   sum((case when a.category = 'act' then b.pvalue end)) as load_act,
         |   (case
         |     when substr(a.point_name, 1, 3) = '00F' then 20
         |     when substr(a.point_name, 1, 3) in ('01F','02F') then 170
         |     else 150
         |   end) as power_rated
         | from
         |   tb_point a left join t_ods_hbase_edos_tmp b on
         |     a.point_code = b.checkPoint
         | where
         |   b.createTime is not null
         | group by
         |   substr(a.point_name, 1, 3),
         |   b.createTime
         | order by
         |   unit_name,
         |   createTime
         |""".stripMargin)
    odsDF.createOrReplaceTempView("t_ods")

    // 从上边结果中，重新整理一下列的位置，处理值为null的列赋默认值0.0
    val etldf = ss.sql(
      s"""
         | select
         |   unit_name,
         |   createTime,
         |   power_rated,
         |   nvl(freq, 0.0) as freq_last,
         |   nvl(power, 0.0) as power_last,
         |   nvl(load_act, 0.0) as load_act,
         |   0.0 as water
         | from
         |   t_ods
         |""".stripMargin)
    etldf.createOrReplaceTempView("t_etl")


    // 计算超过上下限的值
    val etldf2 = ss.sql(
      s"""
         | select
         |   unit_name,
         |   createTime,
         |   power_rated,
         |   freq_last,
         |   power_last,
         |   water,
         |   load_act,
         |   lag(power_last, 1) over(partition by unit_name order by createTime) as power_lag,
         |   round(case when freq_last < $low then ($low-freq_last) when freq_last > $up then ($up-freq_last) end, 4) as diff_f,
         |   (case when freq_last < $low then -1 when freq_last > $up then 1 else 0 end) as beyond_type,
         |   sum(case when freq_last < $low or freq_last > $up then 1 else 0 end) over(partition by unit_name order by createTime) as normal_grp
         | from
         |   t_etl a
         | where
         |   a.freq_last between 49 and 51
         |""".stripMargin)
    etldf2.createOrReplaceTempView("t_etl2")


    //
    val etldf3 = ss.sql(
      s"""
         | select
         |   a.*,
         |   lag(beyond_type, 1) over(partition by unit_name order by createTime) as beyond_type_lag,
         |   (case when diff_f is not null then normal_grp-1 else normal_grp end) as normal_grp2
         | from
         |   t_etl2 a
         |""".stripMargin)
    etldf3.createOrReplaceTempView("t_etl3")


    // frequency_grp = 0表示一直越限
    val etldf4 = ss.sql(
      s"""
         | select
         |   a.*,
         |   sum(case when beyond_type = beyond_type_lag and diff_f is not null then 0 else 1 end) over(partition by unit_name order by createTime) as frequency_grp,
         |   sum(1) over(partition by unit_name, normal_grp2 order by createTime rows between unbounded preceding and current row) as normal_interval
         | from
         |   t_etl3 a
         |""".stripMargin)
    etldf4.createOrReplaceTempView("t_etl4")


    //
    val etldf5 = ss.sql(
      s"""
         | select
         |   a.*,
         |   (case when diff_f is null then frequency_grp-1  else frequency_grp end) as frequency_grp2
         | from
         |   t_etl4 a
         |""".stripMargin)
    etldf5.createOrReplaceTempView("t_etl5")


    /*
    * freq_max --求出最后频率的最大值
    * diff_f_abs_max --求出最后频率 绝对值的最大值
    * freq_min --求出最后频率的最小值
    * power_last_b --返回最后一个功率值
    * power_last_a --返回第一个功率值
    * createTime_b --返回最后创建时间
    * freq_last_b --返回最后频率
    * diff_p_sum --求出一次调频理论贡献电量
    * power_sum --汇总功率值
    * createtime_delay --创建时间 延迟
    * rn2 --求出行数
    * load_act_max --求出负载的最大值
    * power_max --求出功率的最大值
    * power_min --求出功率的最小值
    * */
    val etldf6 = ss.sql(
      s"""
         | select
         |   a.*,
         |   max(freq_last) over(partition by unit_name, frequency_grp2 rows between unbounded preceding and UNBOUNDED FOLLOWING) as freq_max,
         |   max(abs(diff_f)) over(partition by unit_name, frequency_grp2 rows between unbounded preceding and UNBOUNDED FOLLOWING) as diff_f_abs_max,
         |   min(freq_last) over(partition by unit_name, frequency_grp2 rows between unbounded preceding and UNBOUNDED FOLLOWING) as freq_min,
         |   last(power_last, true) over(partition by unit_name, frequency_grp2 order by createTime rows between unbounded preceding and UNBOUNDED FOLLOWING) as power_last_b,
         |   first(power_last, true) over(partition by unit_name, frequency_grp2 order by createTime rows between unbounded preceding and UNBOUNDED FOLLOWING) as power_last_a,
         |   last(createTime, true) over(partition by unit_name, frequency_grp2 order by createTime rows between unbounded preceding and UNBOUNDED FOLLOWING) as createTime_b,
         |   first(freq_last, true) over(partition by unit_name, frequency_grp2 order by createTime rows between unbounded preceding and UNBOUNDED FOLLOWING) as freq_last_a,
         |   last(freq_last, true) over(partition by unit_name, frequency_grp2 order by createTime rows between unbounded preceding and UNBOUNDED FOLLOWING) as freq_last_b,
         |   sum(diff_f*power_rated/(50*$abc)) over(partition by unit_name, frequency_grp2 rows between unbounded preceding and UNBOUNDED FOLLOWING) as diff_p_sum,
         |   sum(power_last) over(partition by unit_name, frequency_grp2 rows between unbounded preceding and UNBOUNDED FOLLOWING) as power_sum,
         |   min(case when beyond_type = beyond_type_lag and (power_last-power_lag)*diff_f > 0 then createTime end) over(partition by unit_name, frequency_grp2 rows between unbounded preceding and UNBOUNDED FOLLOWING) as createtime_delay,
         |   row_number() over(partition by unit_name, frequency_grp2 order by createTime) as rn2,
         |   max(load_act) over(partition by unit_name, frequency_grp2 rows between unbounded preceding and UNBOUNDED FOLLOWING) as load_act_max,
         |   max(power_last) over(partition by a.unit_name, a.frequency_grp2 order by createTime rows between 0 preceding and 5*50 FOLLOWING) as power_max,
         |   min(power_last) over(partition by a.unit_name, a.frequency_grp2 order by createTime rows between 0 preceding and 5*50 FOLLOWING) as power_min
         | from
         |   t_etl5 a
         |""".stripMargin).cache()
    etldf6.createOrReplaceTempView("t_etl6")



    //
    val etldf7 = ss.sql(
      s"""
         | select
         |   a.*,
         |   unix_millis(to_timestamp(a.createTime_b, 'yyyy/MM/dd HH:mm:ss.SSS'))-unix_millis(to_timestamp(a.createTime, 'yyyy/MM/dd HH:mm:ss.SSS')) as during_time,
         |   min(case when power_max-power_min <= power_rated*0.01 then a.createTime end) over(partition by a.unit_name, a.frequency_grp2 rows between unbounded preceding and UNBOUNDED FOLLOWING) as createTime_stable,
         |   unix_millis(to_timestamp(a.createtime_delay, 'yyyy/MM/dd HH:mm:ss.SSS'))-unix_millis(to_timestamp(a.createTime,'yyyy/MM/dd HH:mm:ss.SSS')) as load_delay_time
         | from
         |   t_etl6 a
         |""".stripMargin)
    etldf7.createOrReplaceTempView("t_etl7")


    // 测点表
    val hms_cm_asset = ss.read
      .format("jdbc")
      .option("url", MysqlDml.url2)
      .option("dbtable", "hms_cm_asset")
      .option("user", MysqlDml.user2)
      .option("password", MysqlDml.password2)
      .option("driver", MysqlDml.driver2)
      .load()
    hms_cm_asset.createOrReplaceTempView("hms_cm_asset")


    /** 计算一次调频结果
     * 一次调频各个字段意思：
    * during_time --持续时间
    * start_freq --开始频率
    * end_freq --结束频率
    * beyond_type --越线类型
    * limit_freq --极限频率
    * gen_status --
    * start_power --开始有功
    * end_power --结束有功
    * acu_load_rang --实际负荷调整幅度
    * pri_load_range --理论负荷调整幅度
    * acu_load_power --一次调频贡献电量
    * pri_load_power --一次调频理论贡献电量
    * fmk_value -- 一次调频贡献率K
    * normal --
    * load_act --一次调频动作   动作/未动作
    * load_dir --一次调频方向    正确/不正确
    * load_delay_time --响应滞后时间
    * load_delay_status --响应时间"是否合格: 合格：qualified  不合格：disqualification  不考核
    * act_load_rang_ten_sec --
    * load_ninety_time --
    * act_load_rang_fifteen_sec --
    * stable_time --稳定时间
    * stable_status --稳定时间"是否合格 不考核
    * month_check_up --
    * valid_act --
    * valid_peformance_act --是否为有效调频性能事件 是：yes 否：no
    * check_up_item --项目考核
    * freq_type --频率扰动类型: 小扰动：small-act 大扰动（非阶越）：big-act-normal 大扰动（阶越）：big-act-beyond
    * except_check_up --是否免考核:是：yes 否：no
    * load_correct_result --调频动作正确性考核结果  合格/不合格
    * performance_check_up_result --性能考核结果    合格/不合格
    */
    val resultDf = ss.sql(
        s"""
           | select
           |   c.assetid as locationsid,
           |   createTime as start_time,
           |   createTime_b as end_time,
           |   unix_millis(to_timestamp(createTime, 'yyyy/MM/dd HH:mm:ss.SSS')) as start_time_millis,
           |   unix_millis(to_timestamp(createTime_b, 'yyyy/MM/dd HH:mm:ss.SSS')) as end_time_millis,
           |   during_time/1000 as during_time,
           |   round(freq_last, 4) as start_freq,
           |   round(freq_last_b, 4) as end_freq,
           |   beyond_type,
           |   round((case when beyond_type = 1 then freq_max else freq_min end), 4) as limit_freq,
           |   round(power_last_a, 4) as start_power,
           |   round(power_last_b, 4) as end_power,
           |   round(power_last_b-power_last, 4) as acu_load_rang,
           |   round((abs(freq_last-50)-0.05)*power_rated/(50*$abc), 4) as pri_load_range,
           |   round((power_sum-power_last*(during_time+20)/20)/50, 4) as acu_load_power,
           |   round(diff_p_sum/50, 4) as pri_load_power,
           |   round((power_sum-power_last*(during_time+20)/20)/(diff_p_sum), 4) as fmk_value,
           |   'normal' as load_status,
           |   (case when load_act_max = 1 then '动作' else '未动作' end) as load_act,
           |   (case when (power_sum-power_last*(during_time+20)/20)/(diff_p_sum) > 0 then 'normal' else 'exception' end) as load_dir,
           |   load_delay_time/1000 as load_delay_time,
           |   (case
           |      when diff_f_abs_max > 0 and diff_f_abs_max < 0.03 then '不考核'
           |      when load_delay_time < 10000 then 'qualified'
           |      else 'disqualification'
           |   end) as load_delay_status,
           |   0 as act_load_rang_ten_sec,
           |   0 as load_ninety_time,
           |   0 as act_load_rang_fifteen_sec,
           |   (case
           |      when abs(diff_f) >= 0.03 then (unix_millis(to_timestamp(createTime_stable, 'yyyy/MM/dd HH:mm:ss.SSS'))-unix_millis(to_timestamp(createTime, 'yyyy/MM/dd HH:mm:ss.SSS')))/1000
           |      else 0
           |   end) as stable_time,
           |   (case
           |      when abs(diff_f) >= 0.03 and (unix_millis(to_timestamp(createTime_stable, 'yyyy/MM/dd HH:mm:ss.SSS'))-unix_millis(to_timestamp(createTime, 'yyyy/MM/dd HH:mm:ss.SSS'))) < 45000 then 'qualified'
           |      when abs(diff_f) >= 0.03 then 'disqualification'
           |      else '不考核'
           |   end) as stable_status,
           |   '合格' as month_check_up,
           |   (case
           |      when during_time > $during_time_param1 then 'yes'
           |      else 'no'
           |   end) as valid_act,
           |   (case
           |      when (during_time > $during_time_param2) and ((power_sum-power_last*(during_time+20)/20)/(diff_p_sum) > 0 or load_act_max = 1) and normal_interval*20 > $normal_interval_param1 then 'yes'
           |      else 'no'
           |   end) as valid_peformance_act,
           |   (case
           |      when (power_sum-power_last*(during_time+20)/20)/(diff_p_sum) > 0 and (during_time > $during_time_param1) and normal_interval*20 > $normal_interval_param1 then '性能考核'
           |      else '调频动作正确性考核'
           |   end) as check_up_item,
           |   (case
           |      when diff_f_abs_max > 0 and diff_f_abs_max < 0.03 then 'small-act'
           |      when abs(diff_f) < 0.03 and diff_f_abs_max >= 0.03 then 'big-act-normal'
           |      when abs(diff_f) >= 0.03 then 'big-act-beyond'
           |   end) as freq_type,
           |   (case
           |      when power_last < $power_rated_factor*power_rated then 'yes'
           |      else 'no'
           |   end) as except_check_up,
           |   (case
           |      when (power_sum-power_last*(during_time+20)/20)/(diff_p_sum) > 0 or load_act_max = 1 then '合格'
           |      else '不合格'
           |   end) as load_correct_result,
           |   (case
           |      when diff_f_abs_max > 0 and diff_f_abs_max < 0.03 and abs((power_sum-power_last*(during_time+20)/20)/(diff_p_sum)) > 0.35 then '合格'
           |      when abs(diff_f) < 0.03 and diff_f_abs_max >= 0.03 and abs((power_sum-power_last*(during_time+20)/20)/(diff_p_sum)) > 0.6 and load_delay_time < 10000 then '合格'
           |      when abs(diff_f) >= 0.03 and abs((power_sum-power_last*(during_time+20)/20)/(diff_p_sum)) > 0.6 and load_delay_time < 10000 and (unix_millis(to_timestamp(createTime_stable, 'yyyy/MM/dd HH:mm:ss.SSS'))-unix_millis(to_timestamp(createTime, 'yyyy/MM/dd HH:mm:ss.SSS'))) < 45000 then '合格'
           |      else '不合格'
           |   end) as performance_check_up_result,
           |   'GZB' as project_id,
           |   '10000' as org_id,
           |   '' as sub_org_id,
           |   '' as create_person,
           |   now() as create_time,
           |   '' as modify_person,
           |   now() as modify_time
           | from
           |   t_etl7 a
           | left join hms_cm_asset c on
           |   concat(a.unit_name, '机组') = c.description
           | where
           |   rn2 = 1
           | and
           |   power_last > 10
           | and
           |   diff_f is not null
           |""".stripMargin)
    resultDf.createOrReplaceTempView("kk")


    // 清除旧数据
    val tablename = "hms_eo_fm_interval_pmu"
    val delString = s"delete from $tablename where start_time between '$startTime' and '$endTime'"
    MysqlDml.delete2(delString)

    // 结果写mysql
    resultDf.write
      .format("jdbc")
      .mode("append")
      .option("url", MysqlDml.url2)
      .option("dbtable", tablename)
      .option("user", MysqlDml.user2)
      .option("password", MysqlDml.password2)
      .option("driver", MysqlDml.driver2)
      .save()
  }


  // 一次调频（旧版本：最小维度是秒）
  def calFrequency(ss: SparkSession, sc: SparkContext, hTable: IDaoProvider, time: Array[String], step: Int): Unit = {
    val startTime = time(0)
    val endTime = time(1)

    val sdf = new SimpleDateFormat("yyyy/MM/dd")
    val sdfs = new SimpleDateFormat("yyyy/MM/dd HH:mm:ss")

    // 上下限
    val low = 50 - 0.05
    val up = 50 + 0.05

    val abc = 0.04

    // 读取配置
    val jdbcDF = ss.read
      .format("jdbc")
      .option("url", MysqlDml.url)
      .option("dbtable", "hms_first_frequency_point_conf")
      .option("user", MysqlDml.user)
      .option("password", MysqlDml.password)
      .option("driver", MysqlDml.driver2)
      .load().cache()
    jdbcDF.createOrReplaceTempView("hms_first_frequency_point_conf")

    // 获取机组运行状态
    JzCondition.calData(ss, sc, hTable, time, step, 1)

    import ss.implicits._

    val point_code: Array[String] = jdbcDF.filter("category in ('power','freq','jzwater','act')").map(x => x.getString(0)).collect()

    // 获取测点每一秒的值
    GetData.getBySecond(ss, sc, hTable, point_code, time, step, 1)

    //
    val odsDF = ss.sql("select substr(point_name,1,3) as unit_name, createTime" +
      ",sum(case when a.category='power' then pvalue end) power" +
      ",sum(case when a.category='freq' then pvalue end) freq" +
      ",sum(case when a.category='jzwater' then pvalue end) water" +
      ",sum(case when a.category='act' then pvalue end) load_act" +
      " ,case when substr(point_name,1,3)='00F' THEN 20 when substr(point_name,1,3) in ('01F','02F') then 170 else 150 end power_rated " +
      " from hms_first_frequency_point_conf  a " +
      " left join t_ods_hbase_edos_tmp b on b.checkPoint=a.point_code " +
      " where a.category in ('power','freq','jzwater','act') and createTime is not null" +
      " group by substr(point_name,1,3),createTime")
    odsDF.createOrReplaceTempView("t_ods")
    //    odsDF.filter("unit_name='00F'").orderBy("unit_name","createTime").show()
    //odsDF.orderBy("unit_name","createTime").show(500)
    //odsDF.filter("freq<=40").orderBy("unit_name","createTime").repartition(1).write.csv("F://user/root/first_frequency.csv")
    //odsDF.filter("freq<=40").orderBy("unit_name","createTime").repartition(1).write.csv("/root/gzb/first_frequency.csv")


    //
    val etldf = ss.sql("select unit_name,createTime,power_rated,freq as freq_last,power as power_last,water,load_act" +
      // ",last(power,true) over(partition by unit_name order by createTime) as power_last" +
      " from t_ods a")
    etldf.createOrReplaceTempView("t_etl")
    //    etldf.filter("unit_name='00F'").orderBy("unit_name","createTime").show()


    //
    val etldf2 = ss.sql("select unit_name,createTime,power_rated,freq_last,power_last,water,load_act" +
      ",lag(power_last,1) over(partition by unit_name order by createTime) as power_lag," +
      s"round(case when freq_last<$low then ($low-freq_last) when freq_last>$up then ($up-freq_last) end,4) diff_f ," +
      s"case when freq_last<$low then -1 when freq_last>$up then 1 else 0 end  beyond_type, " +
      s"sum(case when freq_last<$low or freq_last>$up then 1 else 0 end) over(partition by unit_name order by createTime) normal_grp " +
      "from t_etl a " +
      "where a.freq_last between 49 and 51")
    //"where a.freq_last is not null")
    etldf2.createOrReplaceTempView("t_etl2")
    etldf2.filter("unit_name='00F'").orderBy("unit_name", "createTime").show()


    //
    val etldf3 = ss.sql("select a.*" +
      //",power_rated*diff_f+power_last  target " +
      ",lag(beyond_type,1) over(partition by unit_name order by createTime) as beyond_type_lag" +
      ",case when diff_f is not null then normal_grp-1 else normal_grp end normal_grp2 " +
      "from t_etl2 a")
    //etldf3.orderBy("unit_name","createTime").show(500)
    //etldf3.repartition($"unit_name",substring($"createTime",1,10)).cache()
    //    etldf3.filter("unit_name='00F'").orderBy("unit_name","createTime").show()
    //etldf3.filter($"createTime" >"2020/04/04 11:59:00" ).orderBy("unit_name","createTime").show(500)
    etldf3.createOrReplaceTempView("t_etl3")


    //
    val etldf4 = ss.sql("select a.*" +
      s",sum(case when beyond_type=beyond_type_lag and diff_f is not null then 0 else 1 end) over(partition by unit_name order by createTime) frequency_grp " +
      ",sum(1) over(partition by unit_name,normal_grp2 order by createTime rows between unbounded preceding and current row) as normal_interval" +
      " from t_etl3 a  "
    )
    //    etldf4.filter("unit_name='00F'").orderBy("unit_name","createTime").show()
    etldf4.createOrReplaceTempView("t_etl4")
    //etldf4.orderBy("unit_name","createTime","createTime_b").show(500)


    //
    val etldf5 = ss.sql("select a.*" +
      ",case when diff_f is null then frequency_grp-1  else frequency_grp end frequency_grp2 " +
      "from t_etl4 a  "
    )
    etldf5.createOrReplaceTempView("t_etl5")
    //    etldf5.filter("unit_name='00F'").orderBy("unit_name","createTime").show()


    /*
    * freq_max --求出最后频率的最大值
    * diff_f_abs_max --求出最后频率 绝对值的最大值
    * freq_min --求出最后频率的最小值
    * power_last_b --返回最后一个功率值
    * power_last_a --返回第一个功率值
    * createTime_b --返回最后创建时间
    * freq_last_b --返回最后频率
    * diff_p_sum --求出一次调频理论贡献电量
    * power_sum --汇总功率值
    * createtime_delay --创建时间 延迟
    * rn2 --求出行数
    * load_act_max --求出负载的最大值
    * power_max --求出功率的最大值
    * power_min --求出功率的最小值
    * */
    val etldf6 = ss.sql("select a.*" +
      ",max(freq_last) over(partition by unit_name,frequency_grp2 rows between unbounded preceding and UNBOUNDED FOLLOWING) freq_max" +
      s",max(abs(diff_f)) over(partition by unit_name,frequency_grp2 rows between unbounded preceding and UNBOUNDED FOLLOWING) diff_f_abs_max" +
      ",min(freq_last) over(partition by unit_name,frequency_grp2 rows between unbounded preceding and UNBOUNDED FOLLOWING) freq_min" +
      ",last(power_last,true) over(partition by unit_name,frequency_grp2 order by createTime rows between unbounded preceding and UNBOUNDED FOLLOWING) power_last_b" +
      ",first(power_last,true) over(partition by unit_name,frequency_grp2 order by createTime rows between unbounded preceding and UNBOUNDED FOLLOWING) power_last_a" +
      ",last(createTime,true) over(partition by unit_name,frequency_grp2 order by createTime rows between unbounded preceding and UNBOUNDED FOLLOWING) createTime_b" +
      ",last(freq_last,true) over(partition by unit_name,frequency_grp2 order by createTime rows between unbounded preceding and UNBOUNDED FOLLOWING) freq_last_b" +
      s",sum(diff_f*power_rated/(50*$abc)) over(partition by unit_name,frequency_grp2 rows between unbounded preceding and UNBOUNDED FOLLOWING) diff_p_sum" +
      ",sum(power_last) over(partition by unit_name,frequency_grp2 rows between unbounded preceding and UNBOUNDED FOLLOWING) power_sum" +
      ",min(case when beyond_type=beyond_type_lag and (power_last-power_lag)*diff_f>0 then createTime end) over(partition by unit_name,frequency_grp2 rows between unbounded preceding and UNBOUNDED FOLLOWING) createtime_delay" +
      ",row_number() over(partition by unit_name,frequency_grp2 order by createTime) rn2" +
      //增加调频动作最大值 修改功率稳定时间逻辑
      ",max(load_act) over(partition by unit_name,frequency_grp2 rows between unbounded preceding and UNBOUNDED FOLLOWING) load_act_max" +
      ",max(power_last) over(partition by a.unit_name,a.frequency_grp2 order by createTime rows between 0 preceding and 5 FOLLOWING) power_max" +
      ",min(power_last) over(partition by a.unit_name,a.frequency_grp2 order by createTime rows between 0 preceding and 5 FOLLOWING) power_min" +
      " from t_etl5 a ").cache()
    etldf6.createOrReplaceTempView("t_etl6")
    //etldf6.filter("unit_name='02F'").orderBy("unit_name","createTime").show(500)
    //    etldf6.filter("unit_name='00F'").orderBy("unit_name","createTime").show()


    //
    val etldf7 = ss.sql("select a.*" +
      ",unix_timestamp(a.createTime_b,'yyyy/MM/dd HH:mm:ss')-unix_timestamp(a.createTime,'yyyy/MM/dd HH:mm:ss') during_time" +
      ",min(case when power_max-power_min <=power_rated*0.01 then a.createTime end) over(partition by a.unit_name,a.frequency_grp2 rows between unbounded preceding and UNBOUNDED FOLLOWING) createTime_stable" +
      ",unix_timestamp(a.createtime_delay,'yyyy/MM/dd HH:mm:ss')-unix_timestamp(a.createTime,'yyyy/MM/dd HH:mm:ss') load_delay_time" +
      " from t_etl6 a "
    )
    etldf7.createOrReplaceTempView("t_etl7")
    //etldf7.orderBy("unit_name","createTime").show(200)
    //    etldf7.filter("unit_name='00F'").orderBy("unit_name","createTime").show()


    // 测点表
    val hms_cm_asset = ss.read
      .format("jdbc")
      .option("url", MysqlDml.url2)
      .option("dbtable", "hms_cm_asset")
      .option("user", MysqlDml.user2)
      .option("password", MysqlDml.password2)
      .option("driver", MysqlDml.driver2)
      .load()
    hms_cm_asset.createOrReplaceTempView("hms_cm_asset")

    /* 一次调频各个字段意思
    * during_time --持续时间
    * start_freq --开始频率
    * end_freq --结束频率
    * beyond_type --越线类型
    * limit_freq --极限频率
    * gen_status --
    * start_power --开始有功
    * end_power --结束有功
    * acu_load_rang --实际负荷调整幅度
    * pri_load_range --理论负荷调整幅度
    * acu_load_power --一次调频贡献电量
    * pri_load_power --一次调频理论贡献电量
    * fmk_value -- 一次调频贡献率K
    * normal --
    * load_act --一次调频动作   动作/未动作
    * load_dir --一次调频方向    正确/不正确
    * load_delay_time --响应滞后时间
    * load_delay_status --响应时间"是否合格: 合格：qualified  不合格：disqualification  不考核
    * act_load_rang_ten_sec --
    * load_ninety_time --
    * act_load_rang_fifteen_sec --
    * stable_time --稳定时间
    * stable_status --稳定时间"是否合格 不考核
    * month_check_up --
    * valid_act --
    * valid_peformance_act --是否为有效调频性能事件 是：yes 否：no
    * check_up_item --项目考核
    * freq_type --频率扰动类型: 小扰动：small-act 大扰动（非阶越）：big-act-normal 大扰动（阶越）：big-act-beyond
    * except_check_up --是否免考核:是：yes 否：no
    * load_correct_result --调频动作正确性考核结果  合格/不合格
    * performance_check_up_result --性能考核结果    合格/不合格
    * */
    val resultDf = ss.sql("select c.assetid locationsid,water water_head, createTime start_time,createTime_b end_time" +
      //      ",during_time-1 as during_time" + 20210116注释
      ",during_time as during_time" +
      ",round(freq_last,4) start_freq" +
      ",round(freq_last_b,4) end_freq" +
      ",beyond_type" +
      ",round(case when beyond_type=1 then freq_max else freq_min end,4) limit_freq" +
      ",working_name gen_status" +
      // 2021.08.02修改power_last为power_last_a
      ",round(power_last_a,4) start_power" +
      ",round(power_last_b,4) end_power" +
      ",round(power_last_b-power_last,4) acu_load_rang" +
      //      ",round(diff_p_sum,4) pri_load_range" +
      //      ",round(power_sum-power_last*(during_time),4) acu_load_power" +//实际电量
      //      ",round(diff_p_sum*during_time,4) pri_load_power"+//理论电量 diff_p_sum即为理论电量 并不是理论幅度
      //      ",round((power_sum-power_last*(during_time))/(diff_p_sum*(during_time)),4) fmk_value" +
      s",round((abs(freq_last-50)-0.05)*power_rated/(50*$abc),4) pri_load_range" +
      ",round(power_sum-power_last*(during_time+1),4) acu_load_power" +
      ",round(diff_p_sum,4) pri_load_power" +
      ",round((power_sum-power_last*(during_time+1))/(diff_p_sum),4) fmk_value" +
      ",'normal' load_status" +
      ",case when load_act_max=1 then '动作' else '未动作' end  load_act" +
      ",case when (power_sum-power_last*(during_time+1))/(diff_p_sum)>0 then 'normal' else 'exception' end load_dir" +
      ",load_delay_time" +
      ",case when diff_f_abs_max>0 and diff_f_abs_max<0.03 then '不考核' when load_delay_time<10 then 'qualified' else 'disqualification' end load_delay_status" +
      ",0 act_load_rang_ten_sec" +
      ",0 load_ninety_time" +
      ",0 act_load_rang_fifteen_sec" +
      ",case when abs(diff_f)>=0.03 then unix_timestamp(createTime_stable,'yyyy/MM/dd HH:mm:ss')-unix_timestamp(createTime,'yyyy/MM/dd HH:mm:ss') else 0 end stable_time" +
      ",case when abs(diff_f)>=0.03 and unix_timestamp(createTime_stable,'yyyy/MM/dd HH:mm:ss')-unix_timestamp(createTime,'yyyy/MM/dd HH:mm:ss') <45 then 'qualified' when abs(diff_f)>=0.03 then 'disqualification' else '不考核' end stable_status" +
      ",'合格' month_check_up" +
      ",case when during_time-1>0 and nvl(working_name,'机组稳定运行态') in ('机组稳定运行态','机组并网至稳定工况后稳定运行过程','机组并网后增负荷过程') then 'yes' else 'no' end valid_act" +
      ",case when during_time>10 and ((power_sum-power_last*(during_time+1))/(diff_p_sum)>0 or load_act_max=1) and normal_interval>8 then 'yes' else 'no' end valid_peformance_act" +
      ",case when (power_sum-power_last*(during_time+1))/(diff_p_sum)>0 and during_time-1>0 and nvl(working_name,'机组稳定运行态') in ('机组稳定运行态','机组并网至稳定工况后稳定运行过程','机组并网后增负荷过程') and normal_interval>8 then '性能考核' " +
      "else '调频动作正确性考核'  end check_up_item" +
      ",case when diff_f_abs_max>0 and diff_f_abs_max<0.03 then 'small-act' when abs(diff_f)<0.03 and diff_f_abs_max>=0.03 then 'big-act-normal' when abs(diff_f)>=0.03 then 'big-act-beyond' end freq_type" +
      ",case when power_last<0.2*power_rated then 'yes' else 'no' end except_check_up" +
      // ",case when not(during_time-1>0 and nvl(working_name,'机组稳定运行态') in ('机组稳定运行态','机组并网至稳定工况后稳定运行过程','机组并网后增负荷过程')) then '不考核' " +
      ",case when (power_sum-power_last*(during_time+1))/(diff_p_sum)>0 or load_act_max=1 then '合格' else '不合格' end load_correct_result" +
      ",case when diff_f_abs_max>0 and diff_f_abs_max<0.03 and abs((power_sum-power_last*(during_time+1))/(diff_p_sum))>0.35 then '合格'" +
      "      when abs(diff_f)<0.03 and diff_f_abs_max>=0.03 and abs((power_sum-power_last*(during_time+1))/(diff_p_sum))>0.6 and load_delay_time<10 then '合格'" +
      "      when abs(diff_f)>=0.03 and abs((power_sum-power_last*(during_time+1))/(diff_p_sum))>0.6 and load_delay_time<10 and unix_timestamp(createTime_stable,'yyyy/MM/dd HH:mm:ss')-unix_timestamp(createTime,'yyyy/MM/dd HH:mm:ss') <45 then '合格' " +
      "      else '不合格' end performance_check_up_result" +
      ",'GZB' project_id,'10000' org_id,'' sub_org_id,'' create_person,now() create_time,'' modify_person,now() modify_time " +
      " from t_etl7 a " +
      " left join hms_cm_asset c on concat(a.unit_name,'机组')=c.description " +
      " left join t_jzgk_statu e on unit_name=e.crew_code and a.createTime=e.start_time" +
      " where nvl(working_name,'机组稳定运行态') in ('机组稳定运行态','机组并网至稳定工况后稳定运行过程','机组并网后增负荷过程')" +
      " and rn2=1 and diff_f is not null"
    )
    //resultDf.orderBy("unit_name","createTime").show(200)
    resultDf.createOrReplaceTempView("kk")
    resultDf.show(10)


    // 清除旧数据
    val tablename = "hms_eo_fm_interval"
    val delString = s"delete from $tablename where start_time between '$startTime' and '$endTime'"
    MysqlDml.delete2(delString)

    // 结果写mysql
    resultDf.write
      .format("jdbc")
      .mode("append")
      .option("url", MysqlDml.url2)
      .option("dbtable", tablename)
      .option("user", MysqlDml.user2)
      .option("password", MysqlDml.password2)
      .option("driver", MysqlDml.driver2)
      .save()
  }


  /**
   * AGC调节考核最新版
   *
   * @param ss
   * @param sc
   * @param hTable
   * @param time
   * @param step
   */
  def calAgc(ss: SparkSession, sc: SparkContext, hTable: IDaoProvider, time: Array[String], step: Int): Unit = {
    //不要目标值target，按进出死区
    import ss.implicits._
    val startTime = time(0)
    val endTime = time(1)

    // 一些固定參數
    val V0_dj = 0.75
    val T0_dj = 5
    val Pn_dj = 2049
    val ACE_dj = 0
    //agc调节死区变量
    val AGC_dj = 10


    val V0_ej = 0.85
    val T0_ej = 5
    val Pn_ej = 1090
    val ACE_ej = 0
    val AGC_ej = 10
    //改T0+60*abs
    val bl = 60

    // 读mysql
    val jdbcDF = ss.read
      .format("jdbc")
      .option("url", MysqlDml.url)
      .option("dbtable", "hms_first_frequency_point_conf")
      .option("user", MysqlDml.user)
      .option("password", MysqlDml.password)
      .option("driver", MysqlDml.driver2)
      .load().cache()
    jdbcDF.createOrReplaceTempView("hms_first_frequency_point_conf")


    val hms_cm_asset = ss.read
      .format("jdbc")
      .option("url", MysqlDml.url2)
      .option("dbtable", "hms_cm_asset")
      .option("user", MysqlDml.user2)
      .option("password", MysqlDml.password2)
      .option("driver", MysqlDml.driver2)
      .load()
    hms_cm_asset.createOrReplaceTempView("hms_cm_asset")

    val t_asset = ss.sql(
      s"""
         | select
         |	 substr(description, 1, 3) as unit_name,
         |	 sby1 as org_name
         | from
         |	 hms_cm_asset
         | where
         |	 assettype = 'jz'
         |""".stripMargin)
      .cache()
    t_asset.createOrReplaceTempView("t_asset")

    // 测点
    val point_code: Array[String] = jdbcDF.filter("category in ('ygsd','sjyg','water')").map(x => x.getString(0)).collect()

    // 获取测点每一秒的值
    GetData.getBySecond(ss, sc, hTable, point_code, time, step, 1)
    //ss.sql("select * from t_ods_hbase_edos_tmp order by checkPoint,createTime").show(500)

    /*
    * AGC考核取值时保留两位小数但是不进位
    * 原本为
    *   sum(case when a.category='sjyg' then pvalue end) power
    *   sum(case when a.category='ygsd' then pvalue end) power_sd
    * 2021-04-12改为
    * 第一种
    *   round(sum(case when a.category='sjyg' then pvalue end),2,1) power
    *   round(sum(case when a.category='ygsd' then pvalue end),2,1) power_sd
    * 第二种
    *   truncate(sum(case when a.category='sjyg' then pvalue end),2) power
    *   truncate(sum(case when a.category='ygsd' then pvalue end),2) power_sd
    * 第三种
    *   cast(sum(case when a.category='sjyg' then pvalue end) AS decimal(15,2)) as power
    *   cast(sum(case when a.category='ygsd' then pvalue end) AS decimal(15,2)) as power_sd
    * */
    val odsDF: DataFrame = ss.sql(
      s"""
         | select
         | 	substr(point_name, 1, 2) as org_name,
         | 	createTime,
         | 	cast(sum(case when a.category = 'sjyg' then pvalue end) as decimal(15, 2)) as power,
         | 	cast(sum(case when a.category = 'ygsd' then pvalue end) as decimal(15, 2)) as power_sd,
         | 	sum(case when a.category = 'water' then pvalue end) as water_head
         | from
         | 	hms_first_frequency_point_conf a
         | left join t_ods_hbase_edos_tmp b on
         | 	b.checkPoint = a.point_code
         | where
         | 	a.category in ('ygsd', 'sjyg', 'water')
         | 	and createTime is not null
         | group by
         | 	substr(point_name, 1, 2), createTime
         |""".stripMargin).cache()
    odsDF.createOrReplaceTempView("t_ods")
    //odsDF.orderBy("org_name","createTime").show(5000)
    //odsDF.filter("freq<=40").orderBy("unit_name","createTime").repartition(1).write.csv("F://user/root/first_frequency.csv")
    //odsDF.filter("freq<=40").orderBy("unit_name","createTime").repartition(1).write.csv("/root/gzb/first_frequency.csv")

    val etldf = ss.sql(
      s"""
         | select
         | 	org_name,
         | 	createTime,
         | 	power,
         | 	power_sd,
         | 	water_head,
         | 	(case when abs(power-power_sd) > $AGC_dj then 1 else 0 end) as is_agc
         | from
         |	  t_ods a
         |""".stripMargin)
    etldf.createOrReplaceTempView("t_etl")


    val etldf2 = ss.sql(
      s"""
         | select
         |	  org_name,
         |	  createTime,
         |	  power,
         |	  power_sd,
         |	  water_head,
         |	  is_agc,
         |	  lag(is_agc, 1) over(partition by org_name order by createTime) as is_agc_lag,
         |	  lag(power_sd, 1) over(partition by org_name order by createTime) as power_sd_lag
         | from
         |	  t_etl a
         |""".stripMargin)
    etldf2.createOrReplaceTempView("t_etl2")
    //etldf2.orderBy("org_name","createTime").show(500)
    //ss.sql("select * from t_etl2 where createTime between '2020/08/26 06:00:00' and  '2020/08/26 07:10:00' order by org_name,createTime").show(5000)
    //stage8 3-4-6-8 t_ods  stage 5-6-8 t_asset   2-7-8 par_working_condition_identify

    val etldf4 = ss.sql(
      s"""
         | select
         |   a.*,
         |   sum(case when is_agc = is_agc_lag and is_agc = 1 then 0 else 1 end) over(partition by org_name order by createTime) frequency_grp0
         | from
         |   t_etl2 a
         |""".stripMargin)
    etldf4.createOrReplaceTempView("t_etl4")
    //etldf4.orderBy("org_name","createTime").show(200)


    ss.sql(
      s"""
         | select
         |   a.*,
         |   (case when is_agc = 0 then frequency_grp0-1 else frequency_grp0 end) as frequency_grp
         | from
         |   t_etl4 a
         |""".stripMargin)
      .createOrReplaceTempView("t_etl5")

    val etldf6 = ss.sql(
      s"""
         | select
         |   a.*,
         |   last(power, true) over(partition by org_name, frequency_grp order by createTime rows between unbounded preceding and UNBOUNDED FOLLOWING) as power_b,
         |   last(createTime, true) over(partition by org_name, frequency_grp order by createTime rows between unbounded preceding and UNBOUNDED FOLLOWING) as createTime_b,
         |   last(power_sd, true) over(partition by org_name, frequency_grp order by createTime rows between unbounded preceding and UNBOUNDED FOLLOWING) as power_sd_b,
         |   row_number() over(partition by org_name, frequency_grp order by createTime rows between unbounded preceding and current row) as rn,
         |   case when org_name = '大江' then $Pn_dj else $Pn_ej end pn,
         |   case when org_name = '大江' then $V0_dj else $V0_ej end V0,
         |   case when org_name = '大江' then $T0_dj else $T0_ej end T0,
         |   case when org_name = '大江' then $ACE_dj else $ACE_ej end ACE
         | from
         |   t_etl5 a
         |""".stripMargin)
    etldf6.createOrReplaceTempView("t_etl6")
    //etldf6.filter("org_name='大江'").orderBy("org_name","createTime").show(200)


    /*
    * 2021-03-31 改T0+60*abs
    * */
    val etldf7 = ss.sql(
      s"""
         | select
         |   a.org_name,
         |   a.createTime,
         |   a.power,
         |   a.power_sd,
         |   a.water_head,
         |   power_sd_lag,
         |   is_agc,
         |   is_agc_lag,
         |   frequency_grp,
         |   power_b,
         |   createTime_b,
         |   power_sd_b,
         |   rn,
         |   pn,
         |   V0,
         |   T0,
         |   ACE,
         |	 b.power_sd as power_sd3,
         |	 c.power_sd as power_sd4,
         |	 b.power as power3,
         |	 c.power as power4,
         |	(case
         |	  when abs(b.power_sd-power_sd_b) > 0 and abs(b.power_sd-b.power) > 10 and abs(power_sd_b-b.power) < 10 then abs(power_sd_b-power_b)/pn
         |	  when abs(c.power_sd-b.power_sd) > 0 and abs(c.power_sd-c.power) > 10 and abs(b.power_sd-b.power) < 10 and abs(b.power_sd-c.power) < 10 then (abs(power_sd_b-b.power)+abs(power_sd_b-power_b))/pn/2
         |	  else (abs(power_sd_b-b.power)+abs(power_sd_b-power_b)+abs(power_sd_b-c.power))/pn/3
         |	end) as adjust_precision,
         |	(power_b-a.power)*(T0+60*abs(power_sd_b-a.power)/V0)*(power_sd_b-power_b)/abs((power_sd_b-a.power)*(power_sd_b-power_b)*(unix_timestamp(createTime_b, 'yyyy/MM/dd HH:mm:ss')-unix_timestamp(a.createTime, 'yyyy/MM/dd HH:mm:ss'))) as key1_value,
         |	unix_timestamp(createTime_b, 'yyyy/MM/dd HH:mm:ss')-unix_timestamp(a.createTime, 'yyyy/MM/dd HH:mm:ss') as adjust_time,
         |	(case
         |		when (power_b-a.power)*(power_sd_b-a.power) > 0 then '正调节'
         |		when (power_b-a.power)*(power_sd_b-a.power) < 0 then '反调节'
         |	end) as adjust_dir
         |from
         |	t_etl6 a
         |left join t_ods b on
         |	a.org_name = b.org_name
         |	and unix_timestamp(b.createTime, 'yyyy/MM/dd HH:mm:ss') = unix_timestamp(a.createTime_b, 'yyyy/MM/dd HH:mm:ss')+5
         |left join t_ods c on
         |	a.org_name = c.org_name
         |	and unix_timestamp(c.createTime, 'yyyy/MM/dd HH:mm:ss') = unix_timestamp(a.createTime_b, 'yyyy/MM/dd HH:mm:ss')+10
         |where
         |	rn = 1 and is_agc = 1
         |""".stripMargin)
    etldf7.createOrReplaceTempView("t_etl7")
    //etldf7.orderBy("org_name","createTime").show(200)


    val resultDf = ss.sql(
      s"""
         | select
         |   a.org_name as project_name,
         |   water_head,
         |   createTime as start_time,
         |   createTime_b as end_time,
         |   round(power, 4) as start_power,
         |   round(power_b, 4) as end_power,
         |   round(power_sd_lag, 4) as lag_order,
         |   round(power_sd, 4) as start_order,
         |   round(power_sd_b, 4) as end_order,
         |   adjust_time,
         |   adjust_dir,
         |   round((power_b-power)/adjust_time, 4) as adjust_speed,
         |   round(adjust_precision, 4) as adjust_precision,
         |   key1_value,
         |   (case when adjust_precision > 0.01 then 0.01/adjust_precision else 1 end) as key2_value,
         |   (case when adjust_precision > 0.01 then 0.01/adjust_precision else 1 end)*round(key1_value*1, 4) as key_value,
         |   (case
         |      when ((case when adjust_precision > 0.01 then 0.01/adjust_precision else 1 end)*key1_value*1 between 0 and 0.9) then 0
         |      else abs(power_b-power)*(case when adjust_precision > 0.01 then 0.01/adjust_precision else 1 end)*round(key1_value*1, 4)*1*6*ACE
         |   end) as compensate_cost,
         |   'valid' as compensate_valid,
         |   'GZB' as project_id,
         |   '10000' as org_id,
         |   '' as sub_org_id,
         |   '' as create_person,
         |   now() as create_time,
         |   '' as modify_person,
         |   now() as modify_time
         | from
         |   t_etl7 a
         | where
         |   adjust_time > 10 and not (power_sd_lag = power_sd)
         |""".stripMargin)
    //resultDf.filter("org_name='大江'").orderBy("org_name","createTime").show(200)
    // resultDf.show()


    val tablename = "hms_eo_fm_agc"
    val delString = s"delete from $tablename where start_time between '$startTime' and '$endTime'"
    MysqlDml.delete2(delString)

    // 结果写入mysql
    resultDf.write
      .format("jdbc")
      .mode("append")
      .option("url", MysqlDml.url2)
      .option("dbtable", tablename)
      .option("user", MysqlDml.user2)
      .option("password", MysqlDml.password2)
      .option("driver", MysqlDml.driver2)
      .save()
  }

}
